package sunyu.demo.spring.kafka;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.springframework.kafka.support.ProducerListener;
import sunyu.toolkit.core.LogKit;

/**
 * 生产者监听
 *
 * @author 孙宇
 */
public class KafkaProducerListener implements ProducerListener {
    private static final Logger logger = LogKit.getLogger();

    @Override
    public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
        logger.debug("kafka发送数据成功（日志开始）");
        logger.debug("topic:{}", topic);
        logger.debug("key:{}", key);
        logger.debug("value:{}", value);
        logger.debug("partition:{}", partition);
        logger.debug("RecordMetadata:{}", recordMetadata);
        logger.debug("kafka发送数据成功（日志结束）");
    }

    @Override
    public void onError(String topic, Integer partition, Object key, Object value, Exception exception) {
        logger.error("kafka发送数据错误（日志开始）");
        logger.error("topic:{}", topic);
        logger.error("key:{}", key);
        logger.error("value:{}", value);
        logger.error("partition:{}", partition);
        logger.error("Exception:{}", exception);
        logger.error("kafka发送数据错误（日志结束）");
        exception.printStackTrace();
    }

    @Override
    public boolean isInterestedInSuccess() {
        logger.info("kafkaProducer监听器启动");
        return true;
    }
}
